/****

    activequant - activestocks.eu

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License along
    with this program; if not, write to the Free Software Foundation, Inc.,
    51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

	
	contact  : contact@activestocks.eu
    homepage : http://www.activestocks.eu

****/
package org.activequant.data.util.converter;

import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.data.Quote;
import org.activequant.core.domainmodel.data.TradeIndication;
import org.activequant.core.types.TimeFrame;
import org.activequant.data.retrieval.IQuoteSubscriptionSource;
import org.activequant.data.retrieval.ISubscription;
import org.activequant.data.retrieval.ITradeIndicationSubscriptionSource;
import org.activequant.data.retrieval.SubscriptionSourceBase;
import org.activequant.util.pattern.events.IEventListener;

/**
 * Wraps trade indication subscription source and converts it to quote
 * subscription source. Trade indications come into this converter and 
 * quotes leave this converter. Can be used to build a processing chain. <br>
 * Builds quote by setting bid and ask price to the trade price,
 * and bid and ask quantities to that of the trade.<br>
 * Optionally you can specify a spread and bid will be half the spread lower 
 * and ask will be half the spread higher than the trade indication price. 
 * <br/>
 * TradeIndicationToQuoteSubscriptionSourceConverter extends SubscriptionSourceBase&lt;Quote&gt; implements IQuoteSubscriptionSource. Holds the following associated variables:
 * <ul>
 * <li>source(ITradeIndicationSubscriptionSource)</li>
 * <li>spread(double)</li>
 * </ul>
 * Holds the following inherited variables:
 * <ul>
 * <li>activeBackends(Map&lt;Topic, Subscription&gt;)</li>
 * <li>activeSubscriptions(Queue&lt;SubscriptionDelegate&gt;)</li>
 * </ul>
 * <b>History:</b><br> 
 * - [11.11.2007] Created (Mike Kroutikov)<br>
 * - [24.05.2008] Adding possibility to have a spread (Ulrich Staudinger)<br>
 * 
 * @author Mike Kroutikov
 */
public class TradeIndicationToQuoteSubscriptionSourceConverter extends SubscriptionSourceBase<Quote> implements IQuoteSubscriptionSource {
	/**
	 * private final ITradeIndicationSubscriptionSource source;
	 */
	private final ITradeIndicationSubscriptionSource source;
	/**
	 * private double spread = 0.0;
	 */
	private double spread = 0.0; 
	/**
	 * constructs a TradeIndicationToQuoteSubscriptionSourceConverter (extends SubscriptionSourceBase&lt;Quote&gt; implements IQuoteSubscriptionSource)
	 * using the given source(ITradeIndicationSubscriptionSource) to set its associated source(ITradeIndicationSubscriptionSource)
	 * @param source
	 */
	public TradeIndicationToQuoteSubscriptionSourceConverter(ITradeIndicationSubscriptionSource source) {
		super();
		this.source = source;
	}
	/**
	 * constructs a TradeIndicationToQuoteSubscriptionSourceConverter (extends SubscriptionSourceBase&lt;Quote&gt; implements IQuoteSubscriptionSource)
	 * using the given source(ITradeIndicationSubscriptionSource) and spread(double) to set its associated source(ITradeIndicationSubscriptionSource) and spread(double)
	 * @param source
	 * @param spread
	 */
	public TradeIndicationToQuoteSubscriptionSourceConverter(ITradeIndicationSubscriptionSource source, double spread) {
		super();
		this.source = source;
		this.spread = spread; 
	}
	/**
	 * A nested private class. Subscription subclass. Listens for TradeIndications and converts them to Quotes. Holds the following associated variables:
	 * <ul>
	 * <li>tradeSubscription(ISubscription&lt;TradeIndication&gt;)</li>
	 * <li>listener(IEventListener&lt;TradeIndication&gt;)</li>
	 * </ul>
	 * Holds the following inherited variables:
	 * <ul>
	 * <li>isActive(AtomicBoolean)</li>
	 * <li>refCount(AtomicInteger)</li>
	 * <li>instrument(InstrumentSpecification)</li>
	 * <li>event(Event&lt;T&gt;)</li>
	 * </ul>
	 * @author Dimitar
	 *
	 */
	private class QuoteSubscription extends Subscription {
		/**
		 * private final ISubscription&lt;TradeIndication&gt; tradeSubscription;
		 */
		private final ISubscription<TradeIndication> tradeSubscription;
		/**
		 * private final IEventListener&lt;TradeIndication&gt; listener = new IEventListener&lt;TradeIndication&gt;() {...}<br/>
		 * A listener which converts a fired trade(TradeIndication) into a quote(Quote) and then
		 * invokes <code>event.fire(quote)</code> on the inherited event(Event&lt;T&gt;)
		 */
		private final IEventListener<TradeIndication> listener = new IEventListener<TradeIndication>() {
			public void eventFired(TradeIndication trade) {
				log.debug("Trade event received : " + trade);
				Quote quote = convert(trade);
				if(quote != null) {
					fireEvent(quote);
				}
			}
		};
		/**
		 * Constructs a QuoteSubscription (extends Subscription).<br/>
		 * Gets/subscribes an ISubscription&lt;TradeIndication&gt; for the given spec(InstrumentSpecification) and sets its associated tradeSubscription(ISubscription&lt;TradeIndication&gt;) with it.<br/>
		 * Registers the associated listener(IEventListener&lt;TradeIndication&gt;) to listen to events from the associated tradeSubscription(ISubscription&lt;TradeIndication&gt;)
		 * @param spec
		 * @throws Exception
		 */
		private QuoteSubscription(InstrumentSpecification spec) throws Exception {
			this.tradeSubscription = source.subscribe(spec);
			this.tradeSubscription.addEventListener(listener);
		}
		/**
		 * invokes <code>tradeSubscription.activate()</code> on the associated tradeSubscription(ISubscription&lt;TradeIndication&gt;)
		 */
		@Override
		protected void handleActivate() throws Exception {
			tradeSubscription.activate();
		}
		/**
		 * invokes <code>tradeSubscription.cancel()</code> on the associated tradeSubscription(ISubscription&lt;TradeIndication&gt;)
		 */
		@Override
		protected void handleCancel() throws Exception {
			tradeSubscription.cancel();
		}
	}
	/**
	 * returns a Quote by converting the given ti(TradeIndication). The associated spread(double) is used to get the askPrice(= ti.price + spread/2) and bidPrice(= ti.price - spread/2)
	 * @param ti
	 * @return
	 */
	private Quote convert(TradeIndication ti) {
		Quote q = new Quote(ti.getTimeStamp());
		q.setInstrumentSpecification(ti.getInstrumentSpecification());
		q.setAskPrice(ti.getPrice() + spread / 2.);
		q.setBidPrice(ti.getPrice() - spread / 2.);
		q.setBidQuantity(ti.getQuantity());
		q.setAskQuantity(ti.getQuantity());
		
		return q;
	}
	/**
	 * returns the associated source(ITradeIndicationSubscriptionSource)
	 * @return
	 */
	public ITradeIndicationSubscriptionSource getTradeIndicationSource() {
		return source;
	}
	/**
	 * Constructs a QuoteSubscription (extends Subscription) using the given spec(InstrumentSpecification) to initialize it:<br/>
	 * -gets/subscribes an ISubscription&lt;TradeIndication&gt; for the given spec(InstrumentSpecification) and sets its associated tradeSubscription(ISubscription&lt;TradeIndication&gt;) with it.<br/>
	 * -registers the associated listener(IEventListener&lt;TradeIndication&gt;) to listen to events from its associated tradeSubscription(ISubscription&lt;TradeIndication&gt;)
	 */
	@Override
	protected QuoteSubscription createSubscription(InstrumentSpecification spec, TimeFrame timeFrame) {
		try {
			return new QuoteSubscription(spec);
		} catch(Exception ex) {
			log.error(ex);
			ex.printStackTrace();
			throw new RuntimeException(ex);
		}
	}
	/**
	 * Returns/creates an ISubscription&lt;Quote&gt; for the given spec(InstrumentSpecification) and TIMEFRAME_1_TICK.
	 */
	public ISubscription<Quote> subscribe(InstrumentSpecification spec) throws Exception {
		return super.subscribe(spec, TimeFrame.TIMEFRAME_1_TICK);
	}
}
